package com.atguigu.hotitems_analysis

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

/**
 * spark 实现flink功能
 *
 * Project: UserBehaviorAnalysis
 * Package: com.atguigu.hotitems_analysis
 * Version: 1.0
 *
 * Created by  WangJX  on 2019/12/10 21:11
 */
object HotItems {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotItems")

    val path: String = HotItems.getClass.getClassLoader.getResource("UserBehavior.csv").getPath

    val context = new StreamingContext(conf, Seconds(5L))
      .textFileStream(path)
      .map(
        data => {
          val dataArrays: Array[String] = data.split(",")
          (dataArrays(1).trim,
            UserBehavior(dataArrays(0).trim.toLong,
            dataArrays(1).trim.toLong,
            dataArrays(2).trim.toInt,
            dataArrays(3).trim,
            dataArrays(4).trim.toLong))
        }
      )
      .filter(_._2.behavior == "pv")
      .window(Minutes(60), Minutes(5))
      .groupByKey()
  }

}
